/**
 * Copyright ©2003~2019年5月6日 szkingdom
 * @author raohj
 * @date 2019年5月6日
 * @file RBlockQueue.java
 * @comment : please input your comment to this file
 * @version : input version
 * @modify:
 * 		1、 2019年5月6日 由 raohj 新建文件
 */
package com.ccnu.demojava.collection;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * @author raohj
 * @date 2019年5月6日
 * @version
 * @comment 使用经典的生产者/消费者模式测试同步队列
 * 
 *          ArrayBlockingQueue : 数组队列 LinkedBlockingQueue : 链表队列
 */
public class RDelayQueue {
	private static final Logger log = LoggerFactory.getLogger(RDelayQueue.class);

	public static class DelayTask implements Delayed {
		private final int delayMilliseconds;
		private final int trigger;
		private final String msg;

		/**
		 * @param delayMilliseconds 过期时间
		 */
		public DelayTask(String msg, int delayMilliseconds) {
			this.msg = msg;
			this.delayMilliseconds = delayMilliseconds;
			this.trigger = (int) (System.nanoTime()
					+ TimeUnit.NANOSECONDS.convert(delayMilliseconds, TimeUnit.MILLISECONDS));
		}

		@Override
		public int compareTo(Delayed o) {
			DelayTask dt = (DelayTask) o;
			if (this.trigger < dt.trigger) {
				return -1;
			} else if (this.trigger > dt.trigger) {
				return 1;
			}
			return 0;
		}

		/**
		 * 获取当前是否超过延期时间
		 */
		@Override
		public long getDelay(TimeUnit unit) {
			return unit.convert(this.trigger - System.nanoTime(), TimeUnit.MILLISECONDS);
		}
		
		@Override
		public String toString() {
			return this.msg + "|" + this.delayMilliseconds;
		}
	}

	/**
	 * @author raohj
	 * @date 2019年5月6日
	 * @version
	 * @comment 生产者
	 */
	public static class RProduct implements Runnable {
		private BlockingQueue<DelayTask> msgQueue; // 消息队列

		/**
		 * 
		 */
		public RProduct(BlockingQueue<DelayTask> msgQueue) {
			this.msgQueue = msgQueue;
		}

		@Override
		public void run() {
			for (int i = 0; i < 10; i++) {
				DelayTask dt = new DelayTask("Msg" + i, 1000-i);
				try {
					msgQueue.put(dt);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * @author raohj
	 * @date 2019年5月6日
	 * @version
	 * @comment 消费者
	 */
	public static class RConsumer implements Runnable {
		private BlockingQueue<DelayTask> msgQueue; // 消息队列

		/**
		 * 
		 */
		public RConsumer(BlockingQueue<DelayTask> msgQueue) {
			this.msgQueue = msgQueue;
		}

		@Override
		public void run() {
			while (msgQueue.size() > 0) {
				log.info("消费者[" + Thread.currentThread().threadId() + "]:	" + msgQueue.poll().toString());
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}

	public static void test() {
		BlockingQueue<DelayTask> msgQueue = new DelayQueue<DelayTask>();
		ExecutorService es = Executors.newFixedThreadPool(100);
		// 设置10个生产者
		for (int i = 0; i < 10; i++) {
			es.submit(new RProduct(msgQueue));
		}

		// 暂停10s等待产生消息
		try {
			Thread.sleep(3000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

		// 设置5个消费者
		for (int i = 0; i < 5; i++) {
			es.submit(new RConsumer(msgQueue));
		}

		es.shutdown();
	}
}
